package com.lesent.kafka.spring.config;

import org.apache.kafka.common.protocol.types.Field;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author shengtao
 * @Description: kafka的配置
 * @date 2018/12/20 13:48
 */
@Configuration
@EnableKafka
public class KafkaConfig {

    /* kafka工厂配置 */
    @Bean
    ProducerFactory<String,String> producerFactory(){
        Properties properties = new Properties();
        try {
            properties.load(new PathMatchingResourcePatternResolver().getResource("classpath:kafka-producer.properties").getInputStream());
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(0);
        }
        Map<String, Object> map = new HashMap<String, Object>((Map)properties);

        ProducerFactory<String,String> factory = new DefaultKafkaProducerFactory<String, String>(map);
        return factory;
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<String, String>(producerFactory());
    }

    public ConsumerFactory<String,String> consumerFactory(){
        Properties properties = new Properties();
        try {
            properties.load(new PathMatchingResourcePatternResolver().getResource("classpath:kafka-consumer.properties").getInputStream());
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(0);
        }
        Map<String, Object> map = new HashMap<String, Object>((Map)properties);
        ConsumerFactory<String,String> factory = new DefaultKafkaConsumerFactory<String, String>(map);
        return factory;
    }

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String,String> factory =
                new ConcurrentKafkaListenerContainerFactory<String, String>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }
}
